package dataStream;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class ProcessFunc {

    public static class CountWithTimestamp {

        public String key;
        public long count;
        public long lastModified;
    }
    public static void main(String[] args) {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream = env.fromCollection(data);
        dataStream.keyBy(person->person.gender).window(TumblingEventTimeWindows.of(Time.seconds(1000L)));

        // The ProcessFunction can be thought of as a FlatMapFunction with access to keyed state and timers.
        // It handles events by being invoked for each event received in the input stream(s).
        dataStream.keyBy(person -> person.gender).process(new KeyedProcessFunction<String, Person, Object>() {

            /** The state that is maintained by this process function */
            private ValueState<CountWithTimestamp> state;

            // 初始化
            @Override
            public void open(Configuration parameters) throws Exception {
                getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
                super.open(parameters);
            }

            /**
             *
             * @param person
             * @param context  gives access to the element’s event time timestamp, and to the TimerService.
             *                 The TimerService can be used to register callbacks for future event-/processing-time instants.
             * @param collector
             * @throws Exception
             */
            @Override
            public void processElement(Person person, Context context, Collector<Object> collector) throws Exception {

            }

            // With event-time timers, the onTimer(...) method is called when the current watermark is advanced up to or
            // beyond the timestamp of the timer, while with processing-time timers, onTimer(...) is called when wall
            // clock time reaches the specified time. During that call, all states are again scoped to the key with
            // which the timer was created, allowing timers to manipulate keyed state.
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
            }
        });
    }
}
